package org.example.kafka.sendtype;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.example.kafka.common.CommonConstant;
import org.example.kafka.common.KafkaConstants;

import java.util.concurrent.Future;

public class KafkaFutureProducer {
    private static KafkaProducer<String,String> producer = null;

    public static void main(String[] args) {
        producer = new KafkaProducer<String, String>(
                KafkaConstants.producerConfig(StringSerializer.class,
                        StringSerializer.class));
        try{
            ProducerRecord<String,String> record;
            try{
                record = new ProducerRecord<String,String>(CommonConstant.HELLO_TOPIC,"kafkaHello","Olympic 2020");
                Future<RecordMetadata> future = producer.send(record);
                System.out.println("开始乒乓球比赛");
                RecordMetadata recordMetadata = future.get();
                if(null!=recordMetadata){
                    System.out.println("offset:"+recordMetadata.offset()+"----------------"
                            +"partition:"+recordMetadata.partition());
                }

            }catch (Exception e){
                e.printStackTrace();
            }
        }finally {
            producer.close();
        }
    }
}
